chore: disk backpressure class utility #6020

kostasrim · 2025-11-06T11:54:14Z

This PR adds a simple producer/consumer utility class that for a backing file F can:

write/read blobs sequentially from disk

This class will be used by dragonfly connection to offload commands when dispatch queue surpasses a given value.

No functional change in dragonfly_connection -- I just added the initialization paths and the flags needed.

Resolves #6029

Signed-off-by: Kostas Kyrimis <[email protected]>

romange · 2025-11-06T12:03:49Z

src/facade/disk_connection_backpressure.cc

+}
+
+DiskBackedBackpressureQueue::~DiskBackedBackpressureQueue() {
+  auto ec = writer_->Close();


nit: best practices suggest separating between i/o operations and d'tor functions

100% if you recall we had issues around ProtocolClient destructor recently because of that 😄

src/facade/disk_connection_backpressure.h

romange · 2025-11-06T13:57:43Z

src/facade/disk_connection_backpressure.h

+
+#pragma once
+
+#include <util/fibers/uring_file.h>


you do not need uring or linux specific api for this, imho.

There are 3 API's:

Uring (uring_file.h) -> asynchronous fiber blocking

Linux (file.h) -> thread blocking via writev/preadv

C++ files (fstream) - > thread blocking

Why would we block the entire thread on a disk write ?

I meant to use io/file.h interface, you do not need to leak the implementation in h file.
For epoll we can use helio/util/fibers/fiber_file.h though it's not a priority and it's fine to assume this feature is only for iouring enabled systems but in any case you will have to write code in such way that it will build on macos (so better not to leak iouring in header files)

romange · 2025-11-06T13:58:52Z

src/facade/disk_connection_backpressure.h

+
+namespace facade {
+
+class DiskBackedBackpressureQueue {


let's rename it to DiskBackedQueue and the file should be renamed as well.

src/facade/disk_connection_backpressure.h

Copilot

Pull Request Overview

This PR adds disk-backed backpressure support for connections in DragonflyDB. When enabled via the disk_backpressure_offload_watermark flag, connections can offload backpressure data to disk instead of holding everything in memory, helping prevent out-of-memory conditions during high load scenarios.

Introduces DiskBackedBackpressureQueue class for managing disk-backed queue operations
Adds configuration flags for controlling disk backpressure behavior (folder location, file size limits, load batch size)
Integrates disk backpressure support into the Connection class infrastructure

Reviewed Changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 9 comments.

Show a summary per file

File	Description
src/facade/dragonfly_connection.h	Adds forward declaration for `DiskBackedBackpressureQueue` and member variables for disk backpressure support
src/facade/dragonfly_connection.cc	Integrates disk backpressure initialization into Connection constructor with new flag and error handling
src/facade/disk_connection_backpressure.h	Defines new `DiskBackedBackpressureQueue` class interface for disk-backed backpressure management
src/facade/disk_connection_backpressure.cc	Implements disk-backed queue operations including file I/O, offloading, and loading from disk
src/facade/CMakeLists.txt	Adds new disk_connection_backpressure.cc source file to build configuration

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

src/facade/dragonfly_connection.cc

src/facade/disk_connection_backpressure.h

src/facade/disk_connection_backpressure.cc

src/facade/disk_connection_backpressure.h

src/facade/disk_connection_backpressure.cc

src/facade/dragonfly_connection.cc

src/facade/disk_connection_backpressure.h

dranikpg · 2025-11-06T15:16:15Z

src/facade/disk_connection_backpressure.h

+  // For each item loaded from disk it calls f(item) to consume it.
+  // Reads up to max_queue_load_size_ items on each call
+  void LoadFromBacking(std::function<void(io::MutableBytes)> f);


I really dislike such interfaces that internalize action. What if the connection wants to stop based on execution time, some other state conditions, shutdown, etc. It's not the queues job to manage this at all

string_view/size_t Read(std::string* buffer) whould be enough

dranikpg · 2025-11-06T15:17:24Z

src/facade/disk_connection_backpressure.cc

+  auto res = writer_->Write(blob);
+  if (res) {


res looks like a Result to me, but not like an ec -> error_code 🙂

dranikpg · 2025-11-06T15:20:03Z

src/facade/disk_connection_backpressure.cc

+}
+
+size_t DiskBackedBackpressureQueue::TotalInMemoryBytes() const {
+  return offsets_.size() * sizeof(ItemOffset);


sadly there's really no way to get the capacity of a dequeue

src/facade/disk_connection_backpressure.h

Signed-off-by: Kostas Kyrimis <[email protected]>

dranikpg · 2025-11-07T16:26:18Z

src/facade/disk_backed_queue.cc

+
+  next_read_offset_ += bytes.size();
+
+  if (total_backing_items_ > 1) {


can we push and pop at the same time? The variable might have changed

src/facade/disk_backed_queue.cc

src/facade/disk_backed_queue.h

dranikpg · 2025-11-07T16:28:46Z

src/facade/disk_backed_queue.cc

+std::error_code DiskBackedQueue::Pop(std::string* out) {
+  // We read the next item and (if there are more) we also prefetch the next item's size.
+  uint32_t read_sz = next_item_total_bytes_ + (total_backing_items_ > 1 ? sizeof(uint32_t) : 0);
+  buffer.resize(read_sz);


Given that you use buffer here, parallel pull and pop are not allowed. But either way, can't we use out directly to read into?

src/facade/disk_backed_queue.cc

romange

I am holding on with the review, as this class will change (and will become simpler) if we just write raw socket data instead of parsed messages

kostasrim · 2025-11-13T10:24:06Z

@dranikpg we changed direction and we will only push/pop raw data. The rationale behind this is that we want to unblock the ConnFb. See: #5952

romange · 2025-11-13T10:34:51Z

src/facade/disk_backed_queue.h

+  std::unique_ptr<io::ReadonlyFile> reader_;
+
+  size_t total_backing_bytes_ = 0;
+  size_t total_backing_items_ = 0;


total_backing_items_ is redundant now

romange · 2025-11-13T10:36:46Z

src/facade/disk_backed_queue.h

+
+  std::error_code Push(std::string_view blob);
+
+  std::error_code Pop(std::string* out);


it's not a good interface now - as you do not pop an element. You will usually have a destination buffer which you will want to fill till the end of its capacity or less. Similarly to socket Recv() interface.

romange · 2025-11-13T10:38:42Z

src/facade/disk_backed_queue.h

+  // Check if backing file is empty, i.e. backing file has 0 bytes.
+  bool Empty() const;
+
+  std::error_code CloseReader();


why do we need to control Reader and Writer independently?
Seems that if Empty(), you just want to shut the queue altogether

romange · 2025-11-13T10:45:41Z

I think there is still work that needs to be done. In any case, I suggest checking how it will fit once you integreate OnRecv and be able to offload/load inside DflyConn.

chore: disk backpressure class utility

e43e064

Signed-off-by: Kostas Kyrimis <[email protected]>

kostasrim force-pushed the kpr15 branch from d733031 to e43e064 Compare November 6, 2025 11:56

kostasrim requested a review from romange November 6, 2025 12:00

kostasrim self-assigned this Nov 6, 2025

kostasrim marked this pull request as ready for review November 6, 2025 12:00

romange reviewed Nov 6, 2025

View reviewed changes

src/facade/disk_connection_backpressure.h Outdated Show resolved Hide resolved

romange reviewed Nov 6, 2025

View reviewed changes

src/facade/disk_connection_backpressure.h Outdated Show resolved Hide resolved

romange reviewed Nov 6, 2025

View reviewed changes

src/facade/disk_connection_backpressure.h Outdated Show resolved Hide resolved

romange requested a review from Copilot November 6, 2025 14:01

Copilot AI reviewed Nov 6, 2025

View reviewed changes

dranikpg reviewed Nov 6, 2025

View reviewed changes

kostasrim added 4 commits November 7, 2025 17:31

comments

9239369

fixes

e4de405

interface

2f179c8

Signed-off-by: Kostas Kyrimis <[email protected]>

do not close on destructor

bb24e53

kostasrim requested review from dranikpg and romange November 7, 2025 15:51

dranikpg reviewed Nov 7, 2025

View reviewed changes

romange reviewed Nov 9, 2025

View reviewed changes

src/facade/disk_backed_queue.cc Outdated Show resolved Hide resolved

romange reviewed Nov 9, 2025

View reviewed changes

push/pop only raw bytes

c7617ec

kostasrim requested review from dranikpg and romange November 13, 2025 10:22

romange reviewed Nov 13, 2025

View reviewed changes


		next_read_offset_ += bytes.size();

		if (total_backing_items_ > 1) {


		std::error_code Push(std::string_view blob);

		std::error_code Pop(std::string* out);


		#pragma once

		#include <util/fibers/uring_file.h>

chore: disk backpressure class utility #6020

Are you sure you want to change the base?

chore: disk backpressure class utility #6020

Conversation

kostasrim commented Nov 6, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

romange Nov 6, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

romange left a comment

Choose a reason for hiding this comment

Uh oh!

kostasrim commented Nov 13, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

romange Nov 13, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

romange commented Nov 13, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

kostasrim commented Nov 6, 2025 •

edited

Loading

romange Nov 6, 2025 •

edited

Loading

kostasrim commented Nov 13, 2025 •

edited

Loading

romange Nov 13, 2025 •

edited

Loading